package com.atguigu.gmall.item.service.impl;

import com.atguigu.gmall.common.cache.BloomName;
import com.atguigu.gmall.common.cache.GmallCache;
import com.atguigu.gmall.common.constant.RedisConst;
import com.atguigu.gmall.feign.product.SkuInfoFeignClient;
import com.atguigu.gmall.feign.search.SkuEsFeignClient;
import com.atguigu.gmall.item.service.ItemService;
import com.atguigu.gmall.model.product.BaseCategoryView;
import com.atguigu.gmall.model.product.SkuInfo;
import com.atguigu.gmall.model.product.SpuSaleAttr;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author lsj
 * @create 2021-08-10 16:34
 */
@Slf4j
@Service
public class ItemServiceImpl implements ItemService {

    @Autowired
    RedissonClient redissonClient;

    //临时保存数据，本地缓存
    // redis.set(key,value)  Hash O(1)
    // mysql       树   O(logN)
//    private Map<String,Object> cache = new HashMap<>();

    @Autowired
    SkuInfoFeignClient skuInfoFeignClient;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Qualifier("corePool")
    @Autowired
    ThreadPoolExecutor executor;

    @Autowired
    SkuEsFeignClient esFeignClient;


    /**
     * 1、分布式锁版的获取商品详情
     *      1）、锁有自动过期时间，防止死锁
     *      2）、加锁解锁都是原子的
     *      3）、锁的续期  10s，要给他不断续期，业务不中断。锁要自动延长时间？
     *          后台启动一个daemon线程，每隔5秒，让这个锁重新开始倒计时，续期就是直接续满。
     *          new Thread()
     *          new TimerTask(); 10
     *
     *      redisson： 基于redis做的分布式锁，分布式对象
     *
     * 如何做一个
     *   - 可重入、自旋、自动续期、分布式锁
     * redisson：  redis+son
     * redis：中间件
     * redisson：操作redis的客户端， 比：jedis、stringRedisTemplate强大
     * @param skuId
     * @return
     */
    @SneakyThrows  //自己设置的使用原生redis操作实现的分布式可重入锁业务
    public Map<String, Object> getSkuInfoWithRedisLock01(Long skuId,Boolean locked,String beforetToken) {

        // while ()  不能上来就用锁
        //1、先判断缓存中是否存在 单位时间内一个人的所有都执行完了
        System.out.println("Thread.currentThread() = " + Thread.currentThread());
        Map<String, Object> cache = queryFromCache(skuId);
        if (cache != null) {
            //2、缓存中有用缓存的
            log.info("缓存命中.....");
            return cache;
        } else {
            //3、缓存中没有调用业务逻辑真正查询
            //3.1）、为了不全放给数据库，占锁来查数据库
            log.info("缓存不命中，开始抢锁.....");
            //第一次连接告诉redis，占坑  用 "lock"
            //底层调用 set lock 1 NX EX 20
            String token =  beforetToken;
            Boolean aBoolean = locked ;
            if(!locked){  //判断之前有人已经帮我锁定了，我就直接用别人的锁
                token = UUID.randomUUID().toString();  //自己加锁就是新token
                //重入锁的设计
                aBoolean = stringRedisTemplate.opsForValue().setIfAbsent("lock", token,20, TimeUnit.SECONDS);
            }
            ;
            //还没运行断电了, 不可重入锁会死锁。  未来设计的所有锁都应该是可重入
            if (aBoolean) {
                //第二次 才设置过期时间
//                stringRedisTemplate.expire("lock",20, TimeUnit.SECONDS);
                //缓存中没有，并且占锁成功才执行业务
                //3.2）、占锁成功
                try {
                    log.info("抢占成功.....");
                    cache = getFromServiceItemFeign(skuId);
//                Thread   sleep是线程阻塞操作
                    //4、查询到数据后放入缓存
                    saveToCache(cache, skuId);  //异常
                } finally {
                    //5、删除锁，一定得执行
                    //保证业务正常出现了异常，finally兜底
                    //断电这种故障，redis过期时间,让redis自己删除\
                    //下面的删锁代码必须是原子型，否则可能出问题
//                    String lock = stringRedisTemplate.opsForValue().get("lock");
//                    if(token.equals(lock)){
//                        //删除我自己的锁
//                        stringRedisTemplate.delete("lock");
//                        System.out.println("删除分布式锁....");
//                        log.info("删除分布式锁....");
//                    }

                    //告诉redis，看 KEYS 是否等于 ARGV（argValue） ，如果是就删除 KEYS，否则返回0
                    String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
                            "then return redis.call('del', KEYS[1]) else return 0 end";
                    //lua脚本原子执行
                    stringRedisTemplate.execute(new DefaultRedisScript<>(script), Arrays.asList("lock"),token);

                }
//                try{
//                    //业务代码  1000
//                }finally {
//                    stringRedisTemplate.delete("lock");
//                }

            } else {

                //3.3）、占锁失败
                //??? 不断的操作redis    非公平锁，没有占锁成功就一直抢占
                // stringRedisTemplate.opsForValue().setIfAbsent("lock", "aaa")
                // stringRedisTemplate.opsForValue().setIfAbsent("lock", "aaa") == true  //生产的写法
                //stringRedisTemplate.opsForValue().setIfAbsent("lock", "aaa") == false
                // 一直抢直到占到锁，写死while(true)
                log.info("没抢成功，开始尝试自旋占锁.....");
                while (true) {
                    log.info("自旋中.....");
                    Thread.sleep(200);  //同一线程里面锁应该直接使用
                    String token2 = UUID.randomUUID().toString();
                    Boolean absent = stringRedisTemplate.opsForValue().setIfAbsent("lock", token2);
                    if(absent){
                        log.info("自旋占锁成功....."); //while(true) 自己调用自己，立马栈溢出
                        //自旋锁成功才调用自己, 栈溢出的写法，但是能看出效果
                        //这个方法能进去抢占成功，是因为absent已经超时了，redis删了
                        //锁设计为可重入锁
                        cache = getSkuInfoWithRedisLock01(skuId,absent,token2);
                        //释放锁  cpu速度远大于 redis，请用原子删锁和原子加锁

                        String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
                                "then return redis.call('del', KEYS[1]) else return 0 end";
                        //lua脚本原子执行
                        stringRedisTemplate.execute(new DefaultRedisScript<>(script), Arrays.asList("lock"),token2);
                        return cache;
                    }
                }
            }
        }

        return cache;
    }


    @SneakyThrows
    void saveToCache(Map<String, Object> data, Long skuId) {
        ObjectMapper objectMapper = new ObjectMapper();
        String string = objectMapper.writeValueAsString(data);
        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        //null值也要缓存
        operations.set("sku:info:" + skuId, string);
    }

    /**
     * 查缓存
     *
     * @param skuId
     * @return
     */
    @SneakyThrows
    public Map<String, Object> queryFromCache(Long skuId) {
        ObjectMapper mapper = new ObjectMapper();
        ValueOperations<String, String> operations =
                stringRedisTemplate.opsForValue();
        String redisContent = operations.get("sku:info:" + skuId);
        if (StringUtils.isEmpty(redisContent)) {
            return null;
        } else {
            return mapper.readValue(redisContent, new TypeReference<Map<String, Object>>() {
            });
        }
    }


    /**
     * 1、redisson会死锁吗？
     *      不会
     *      1、redisson 是可重入锁（同一线程内前面加过后面不用加）
     *      2、redisson 会自动解锁吗？
     *         会：看门狗30s。代码闪断，定时线程没有，就不自动续期了，等30s，redis自己删除
     *         默认加锁就是30s
     *      3、推荐不要使用 lock(30,TimeUnit.SECONDS)；失去了自动续期功能、
     *        scheduleExpirationRenewal(threadId); 的调用前提 是 if (leaseTime == -1)
     *        没有续期功能;
     *
     * 看门狗为什么？只是为了业务超时，锁时间不够来不停续期的
     *  看门狗还有默认的30秒（lock.lock()）自动过期（redis删），就是防止特殊情况
     *  类似断电 可以正常 释放锁 避免造成死锁吗
     *
     * 看门狗是续期的。每隔10s续满期。
     *
     *
     *
     * 2、redisson分布式业务超时怎么办？
     *      1、看门狗自动续期，每隔1/3看门狗时间就续满看门狗时间
     *
     * @param skuId
     * @return
     */
    @SneakyThrows  //使用redisson分布式锁做的方法
    public Map<String, Object> getSkuInfoWithRedissonDistributeLock(Long skuId) {


        log.info("准备查询"+skuId+" 号数据");
//        Lock lock1 = new ReentrantLock();
//        lock1.lock();
        //这段代码是加锁的吗？
        RLock lock = redissonClient.getLock("lock"); //把锁对象拿到
        /**
         * //1、原生方式
         * locked: 默认传false
         * beforeToken默认传null
         */
//        return getSkuInfoWithRedisLock01(skuId,false,null);
//        RFuture<Void> voidRFuture = lock.lockAsync(); //异步加锁
//        System.out.println("aaa");
//
//        //监听，异步回调
//        voidRFuture.onComplete((v,e)->{
//            //业务代码
//            System.out.println("bbb");
//        });

        //2、redisson版
        Map<String, Object> cache = queryFromCache(skuId);
        if(cache == null){
            log.info("redis缓存没命中...准备查询数据库...尝试加锁");
            //3、加锁
            try{
                //看门狗，用来自动续期,redisson的所有功能都是原子性的
                lock.lock();  //这是一个自旋,可以续期，锁的默认时间是30 * 1000; 30s
                //每隔10s定时任务会自动续满期 internalLockLeaseTime / 3，
//                lock.lock(30,TimeUnit.SECONDS);  //30秒以后自动解锁，不会自动续期
                log.info("加锁成功....尝试继续命中缓存....");
                Map<String, Object> cacheAgain = queryFromCache(skuId);
                if(cacheAgain == null){
                    log.info("开始查询数据库....");
                    Thread.sleep(300);
                    Map<String, Object> data = getFromServiceItemFeign(skuId);
                    saveToCache(data,skuId);
                    return data;
                }

                log.info("命中缓存，直接返回....");
                return cacheAgain;
            }finally {
                lock.unlock(); //解锁代码由于前面断电？
            }
        }
        log.info("缓存命中，直接返回.....");
        return cache;
    }

    /**
     * 如果某个订单模块的集群超过数据库并发数，是不是需要分布式锁？那前面需要再加上【布隆过滤器】吗？
     * 1、有缓存就加 【布隆过滤器】维护布隆过滤器也不容易，【每天晚上定时任务重建布隆过滤器】
     * 2、数据库并发数（....），
     *    各种业务应该有自己的布隆过滤器，每个布隆都很小，效率很高
     * AOP+注解完成的
     *    分布式布隆过滤防击穿高性能本地锁数据缓存切面
     *
     * 缓存空值
     * @param skuId
     * @return
     */
    @GmallCache(
            cacheKeyExpr = RedisConst.SKUKEY_PREFIX+"#{#args[0]}"+RedisConst.SKUKEY_SUFFIX,
            ttl = 1000 * 60 * 30,
            bloomName = BloomName.SKU, //用哪个布隆过滤器可以自己指定
            missDataTtl = 1000 * 60 * 10
    )  //需要一个切面类，动态切入所有标了这个注解的方法
    @Override  //sku:51:info
    public Map<String, Object> getSkuInfo(Long skuId) {
        //查数据
        log.info("要查数据库了.......");
        Map<String, Object> map = getFromServiceItemFeign(skuId);
        return map;
    }


    /**
     * 热度延迟更新
     * @param skuId
     */
    @Override
    public void updateSkuHotScore(Long skuId) {
        //每次点击商品都更新热度
        // 100w    100  当redis  skuId :  200
        CompletableFuture.runAsync(()->{
            ZSetOperations<String, String> zset = stringRedisTemplate.opsForZSet();
            //只是给redis里面加值
            Double hotScore = zset.incrementScore("hotScore", skuId.toString(), 1);
            if(hotScore % 10 == 0){
                esFeignClient.updateSkuHotScore(skuId,hotScore.longValue());
            }
        },executor);
    }

    /**
     * 远程调用 service-product 查询出和当前 skuId 对应的所有信息
     * 缓存的使用逻辑是固定，我们完全可以抽取
     * <p>
     * 就可以使用AOP直接抽取出来,数据改了，缓存也要改？  如何使用SpringCache简化缓存开发
     *
     * @param skuId
     * @return 1、百万并发进来，正好skuId=50号数据过期
     * <p>
     * synchronized: 什么锁？ 对象锁，这个锁用的是 ItemServiceImpl 对象，this
     * 只要所有人啥都共用一把锁就能锁住
     * 能：因为 ItemServiceImpl 单实例的。大家进来都是同一个，会大，资源也有消耗
     * service有一个峰值： 4000/s
     * <p>
     * juc的所有锁都是本地锁，只针对当前应用有效，分布式情况锁不住
     */

//    @Override
//    public Map<String, Object> getSkuInfo(Long skuId) {
//////        new xxx()
//////        new Thread(xxxx)
//////        new Thread(xxxx)
//////        new Thread(xxxx)
//////        new Thread(xxxx)
////        //只需要让所有人用同一把锁就能锁住
////        //场景： 数据库MySQL：   修改一条数据   1000   update xxx where id=5
////        //锁可以是数据库。
////        //大家都去一个地方占位，能占到说明拿到锁。这个位有东西，那就说明别人占用了锁，我们就稍等
////        //最快的就是redis
////        //1、test_lock    id   value
////        //                1    xxxxx
////        //2、minio。存一个同名文件。只要我存进来，别人一看有就不存了。
////        //3、分布式中间件，谁都能占坑。mq。create queue。
//////        synchronized (ItemServiceImpl.class){
//////            //效果一样,Spring原因，this已经是单例锁了。
//////            //同一个jvm上部署
//////            //分布式下。应用在多个机器
//////        }
////        ObjectMapper mapper = new ObjectMapper();
////        /**
////         * 1、所有查询之前，先看缓存
////         *   1.1）、如果缓存中有就使用缓存的
////         *   1.2）、如果缓存没有，就查数据库
////         */
////        ValueOperations<String, String> operations =
////                stringRedisTemplate.opsForValue();
////
////        //  50 51  52   sku:info:51    sku:info:52   sku:info:53
////        //1、先去缓存确定是否存在
////        String redisContent = operations.get("sku:info:" + skuId);
////
////
////        if (StringUtils.isEmpty(redisContent)) {
////            //分布式锁
////            // String   ok =   set a b NX
//////            if(ok){
//////                //缓存中没有，远程查询
//////                Map<String, Object> fromServiceItemFeign = getFromServiceItemFeign(skuId);
//////            }else{
//////              //自旋锁
////            // while((set a b NX).equals("ok")){
////            //      getSkuInfo(50);
////            // }
//////            }
////            //2、缓存中没有，远程查询
////            Map<String, Object> fromServiceItemFeign = getFromServiceItemFeign(skuId);
////            //请存到磁盘，存到远程？？？？
//////            ObjectOutputStream objectOutputStream = new ObjectOutputStream();
//////            //java的序列化
//////            objectOutputStream.writeObject(fromServiceItemFeign);
////
////            String jsonStr = null;
////            try {
////                //序列化......
////                jsonStr = mapper.writeValueAsString(fromServiceItemFeign);
////            } catch (JsonProcessingException e) {
////                e.printStackTrace();
////            }
////            //2.1）、给redis中存一份
////            operations.set("sku:info:" + skuId, jsonStr);
////
////            //2.2）、返回数据
////            return fromServiceItemFeign;
////        } else {
////            Map<String, Object> stringObjectMap = null;
////            try {
////                //redis拿到的是str还要转对象   反序列化
////                stringObjectMap = mapper.readValue(redisContent, new TypeReference<Map<String, Object>>() {
////                });
////            } catch (JsonProcessingException e) {
////                e.printStackTrace();
////            }
////            return stringObjectMap;
////        }
//
//        //把redis锁释放
//    }


    private Map<String, Object> getFromServiceItemFeign(Long skuId) {
        Map<String, Object> result = new HashMap<>();
        //我以下的写法，
        // 对比  new Thread(()->{}).start(); 优点：
        //1、线程重复使用。 8~16直接重复使用，少了开线程时间  new Thread  0.01
        //2、线程吞吐量的限制。
        // 以前 1个请求 5个线程，   100请求，new 500个线程   1w个请求  5w线程等待CPU切换（内存占用更大）
        // 现在 1个请求 5个线程，交给线程池，线程池只有16个一直执行。控制资源
        //     100 请求 500个线程，交给线程池， 之前16个线程等待CPU切换，  484 就在队列等待执行
        //     1w 请求  1w个线程，交给线程池，之前16个线程等待CPU切换，  9984 个在队列（占内存）

        //1、查询sku详情
        CompletableFuture<SkuInfo> future = CompletableFuture.supplyAsync(() -> {
            SkuInfo skuInfo = skuInfoFeignClient.getSkuInfo(skuId);
            result.put("skuInfo", skuInfo);
            return skuInfo;
        }, executor);


        //1~4:可交给线程池并行执行
        //1、查询分类
        CompletableFuture<Void> future1 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                BaseCategoryView skuCategorys = skuInfoFeignClient.getSkuCategorys(res.getCategory3Id());
                result.put("categoryView", skuCategorys);
            }
        }, executor);


        //2、查询销售属性
        CompletableFuture<Void> future2 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                List<SpuSaleAttr> spuSaleAttrListCheckBySku = skuInfoFeignClient.getSpuSaleAttrListCheckBySku(skuId, res.getSpuId());
                result.put("spuSaleAttrList", spuSaleAttrListCheckBySku);
            }
        }, executor);

        //3、查询价格
        CompletableFuture<Void> future3 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                BigDecimal skuPrice = skuInfoFeignClient.getSkuPrice(skuId);
                result.put("price", skuPrice);
            }
        }, executor);

        //4、查询组合
        CompletableFuture<Void> future4 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                Map map = skuInfoFeignClient.getSkuValueIdsMap(res.getSpuId());
                ObjectMapper mapper = new ObjectMapper();
                try {
                    String jsonStr = mapper.writeValueAsString(map);
                    log.info("valuesSkuJson 内容：{}", jsonStr);
                    result.put("valuesSkuJson", jsonStr);
                } catch (JsonProcessingException e) {
                    log.error("商品sku组合数据转换异常：{}", e);
                }
            }
        }, executor);


        CompletableFuture<Void> allOf = CompletableFuture.allOf(future, future1, future2, future3, future4);

        try {
            allOf.get();
        } catch (Exception e) {
            log.error("线程池异常：{}",e);
        }
        return result;
    }

    /**
     * 远程查询sku详细信息
     *
     * @param skuId
     * @return
     */
    private Map<String, Object> getFromServiceItemFeign01(Long skuId) {
        log.info("开始远程查询，远程会操作数据库.....");

        Map<String, Object> result = new HashMap<>();
        //skuInfo信息

        //RPC 查询  skuDetail
        //1、Sku基本信息（名字，id，xxx，价格，sku_描述） sku_info
        //2，Sku图片信息（sku的默认图片[sku_info]，sku_image[一组图片]）
        SkuInfo skuInfo = skuInfoFeignClient.getSkuInfo(skuId);
        if (skuInfo != null) {
            result.put("skuInfo", skuInfo);
            //3，Sku分类信息（sku_info[只有三级分类]，根据这个三级分类查出所在的一级，二级分类内容，连上三张分类表继续查）
            BaseCategoryView skuCategorys = skuInfoFeignClient.getSkuCategorys(skuInfo.getCategory3Id());
            result.put("categoryView", skuCategorys);
            //4，Sku销售属性相关信息（查出自己的sku组合，还要查出这个sku所在的spu定义了的所有销售属性和属性值）
            List<SpuSaleAttr> spuSaleAttrListCheckBySku = skuInfoFeignClient.getSpuSaleAttrListCheckBySku(skuId, skuInfo.getSpuId());
            result.put("spuSaleAttrList", spuSaleAttrListCheckBySku);

            //5，Sku价格信息（平台可以单独修改价格，sku后续会放入缓存，为了回显最新价格，所以单独获取）
            BigDecimal skuPrice = skuInfoFeignClient.getSkuPrice(skuId);
            result.put("price", skuPrice);
            //6、SPU下面的所有存在的sku组合信息 {"121|123|156":65,"122|123|111":67}
            //前端这里还需要把map转成json字符串， JSON.parse(.valuesSkuJson)
            Map map = skuInfoFeignClient.getSkuValueIdsMap(skuInfo.getSpuId());
            ObjectMapper mapper = new ObjectMapper();
            try {
                String jsonStr = mapper.writeValueAsString(map);
                log.info("valuesSkuJson 内容：{}", jsonStr);
                result.put("valuesSkuJson", jsonStr);
            } catch (JsonProcessingException e) {
                log.error("商品sku组合数据转换异常：{}", e);
            }

        }
        return result;
    }

}
